/**
 * Copyright 2009 Erick Armbrust 
 * Licensed under the Apache License, Version 2.0 (the "License"); 
 * you may not use this file except in compliance with the License. 
 * You may obtain a copy of the License at 
 * 
 *   http://www.apache.org/licenses/LICENSE-2.0 
 *   
 * Unless required by applicable law or agreed to in writing, software 
 * distributed under the License is distributed on an "AS IS" BASIS, 
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 
 * See the License for the specific language governing permissions and 
 * limitations under the License. 
 */

package tapioca.util;

import java.util.ConcurrentModificationException;
import java.util.List;


import com.google.appengine.api.datastore.DatastoreFailureException;
import com.google.appengine.api.datastore.DatastoreService;
import com.google.appengine.api.datastore.DatastoreTimeoutException;
import com.google.appengine.api.datastore.Key;
import com.google.appengine.api.datastore.Transaction;
import com.google.appengine.repackaged.com.google.common.collect.Lists;

public abstract class EntityGroupBatchJob<T> {
  
  protected final DatastoreService ds;
  protected List<T> toProcess;
  
  public EntityGroupBatchJob(DatastoreService ds, List<T> toProcess) {
    this.ds = ds;
    this.toProcess = toProcess;
  }
  
  public abstract Key getKeyFromItem(T item);
  
  public abstract void processItem(Transaction txn, T item);
  
  // TODO (earmbrust): Switch to processing an entire entity group for efficiency.
  //public abstract void processEntityGroup(Transaction txn, List<T> items);
  
  /**
   * Runs the job by batching all calls to single entity groups into a single transaction.  Every
   * time a successful entity group transaction is executed, the checkpoint is updated.  If any
   * given entity group transaction fails, the loop returns to the checkpoint and restarts 
   * processing.  
   */
  public final void execute() {
    Transaction txn = ds.beginTransaction();
    Key curEntityGroup = null;
    int checkpoint = 0;
    Boolean retry = null;
    
    while (retry == null || retry == true) {
      // At the beginning of every loop, assume that we will succeed and that a retry is unneeded.
      retry = false;
      
      try {
        List<T> entityGroup = Lists.newArrayList();
        for (int i = checkpoint; i < toProcess.size(); i++) {
          T item = toProcess.get(i);
          Key putRequestEntityGroup = KeyUtil.getRootKey(getKeyFromItem(item));
          if (curEntityGroup == null) {
            curEntityGroup = putRequestEntityGroup;
            
          // We have encountered a new entity group, so commit the curren transaction and start
          // a new one.
          } else if (!KeyUtil.inSameEntityGroup(curEntityGroup, putRequestEntityGroup)) {
            if (txn.isActive()) {
              txn.commit();
              txn = ds.beginTransaction();
              checkpoint = i;
            }
            curEntityGroup = putRequestEntityGroup;
          } else {
            entityGroup.add(item);
          }
          
          processItem(txn, item);
        }
        
        // This is needed to commit the final transaction in any given loop.
        if (txn.isActive()) {
          txn.commit();
        }
        
      // These are the only "acceptable" retry exceptions currently.  Any other RuntimeException
      // such as DatastoreNeedIndexException are not valid for retries.
      // TODO (earmbrust): Check with maxr and ryanb to see if this is correct.
      } catch (ConcurrentModificationException e) {
        retry = true;
      } catch (DatastoreTimeoutException e) {
        retry = true;
      } catch (DatastoreFailureException e) {
        retry = true;
      } finally {
        // If the current transaction is active, this is a transaction that did not succeed so it
        // should be rolled back.
        if (txn.isActive()) {
          txn.rollback();
        }
        
        // If we are going to loop around and retry, we are going to need a fresh transaction.
        if (retry) {
          txn = ds.beginTransaction();
        }
      }
    }
  }
  
}
